/*
 * Copyright 2012-2020 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.springframework.cloud.loadbalancer.blocking.client;

import java.io.IOException;
import java.net.URI;
import java.util.Set;

import reactor.core.publisher.Mono;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.loadbalancer.CompletionContext;
import org.springframework.cloud.client.loadbalancer.DefaultRequest;
import org.springframework.cloud.client.loadbalancer.DefaultRequestContext;
import org.springframework.cloud.client.loadbalancer.DefaultResponse;
import org.springframework.cloud.client.loadbalancer.EmptyResponse;
import org.springframework.cloud.client.loadbalancer.HttpRequestLoadBalancerRequest;
import org.springframework.cloud.client.loadbalancer.LoadBalancerClient;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycle;
import org.springframework.cloud.client.loadbalancer.LoadBalancerLifecycleValidator;
import org.springframework.cloud.client.loadbalancer.LoadBalancerProperties;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequest;
import org.springframework.cloud.client.loadbalancer.LoadBalancerRequestAdapter;
import org.springframework.cloud.client.loadbalancer.LoadBalancerUriTools;
import org.springframework.cloud.client.loadbalancer.Request;
import org.springframework.cloud.client.loadbalancer.RequestData;
import org.springframework.cloud.client.loadbalancer.RequestDataContext;
import org.springframework.cloud.client.loadbalancer.Response;
import org.springframework.cloud.client.loadbalancer.ResponseData;
import org.springframework.cloud.client.loadbalancer.TimedRequestContext;
import org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer;
import org.springframework.http.HttpRequest;
import org.springframework.http.client.ClientHttpResponse;
import org.springframework.util.ReflectionUtils;

import static org.springframework.cloud.client.loadbalancer.reactive.ReactiveLoadBalancer.REQUEST;

/**
 * The default {@link LoadBalancerClient} implementation.
 *
 * @author Olga Maciaszek-Sharma
 * @since 2.2.0
 */
@SuppressWarnings({ "unchecked", "rawtypes" })
public class BlockingLoadBalancerClient implements LoadBalancerClient {

	private final ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory;

	public BlockingLoadBalancerClient(ReactiveLoadBalancer.Factory<ServiceInstance> loadBalancerClientFactory) {
		this.loadBalancerClientFactory = loadBalancerClientFactory;
	}

	@Override
	public <T> T execute(String serviceId, LoadBalancerRequest<T> request) throws IOException {
		String hint = getHint(serviceId);
		LoadBalancerRequestAdapter<T, TimedRequestContext> lbRequest = new LoadBalancerRequestAdapter<>(request,
				buildRequestContext(request, hint));
		Set<LoadBalancerLifecycle> supportedLifecycleProcessors = getSupportedLifecycleProcessors(serviceId);
		supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onStart(lbRequest));
		ServiceInstance serviceInstance = choose(serviceId, lbRequest);
		if (serviceInstance == null) {
			supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
					new CompletionContext<>(CompletionContext.Status.DISCARD, lbRequest, new EmptyResponse())));
			throw new IllegalStateException("No instances available for " + serviceId);
		}
		return execute(serviceId, serviceInstance, lbRequest);
	}

	private <T> TimedRequestContext buildRequestContext(LoadBalancerRequest<T> delegate, String hint) {
		if (delegate instanceof HttpRequestLoadBalancerRequest) {
			HttpRequest request = ((HttpRequestLoadBalancerRequest) delegate).getHttpRequest();
			if (request != null) {
				RequestData requestData = new RequestData(request);
				return new RequestDataContext(requestData, hint);
			}
		}
		return new DefaultRequestContext(delegate, hint);
	}

	@Override
	public <T> T execute(String serviceId, ServiceInstance serviceInstance, LoadBalancerRequest<T> request)
			throws IOException {
		if (serviceInstance == null) {
			throw new IllegalArgumentException("Service Instance cannot be null");
		}
		DefaultResponse defaultResponse = new DefaultResponse(serviceInstance);
		Set<LoadBalancerLifecycle> supportedLifecycleProcessors = getSupportedLifecycleProcessors(serviceId);
		Request lbRequest = request instanceof Request ? (Request) request : new DefaultRequest<>();
		supportedLifecycleProcessors
				.forEach(lifecycle -> lifecycle.onStartRequest(lbRequest, new DefaultResponse(serviceInstance)));
		try {
			T response = request.apply(serviceInstance);
			Object clientResponse = getClientResponse(response);
			supportedLifecycleProcessors
					.forEach(lifecycle -> lifecycle.onComplete(new CompletionContext<>(CompletionContext.Status.SUCCESS,
							lbRequest, defaultResponse, clientResponse)));
			return response;
		}
		catch (IOException iOException) {
			supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
					new CompletionContext<>(CompletionContext.Status.FAILED, iOException, lbRequest, defaultResponse)));
			throw iOException;
		}
		catch (Exception exception) {
			supportedLifecycleProcessors.forEach(lifecycle -> lifecycle.onComplete(
					new CompletionContext<>(CompletionContext.Status.FAILED, exception, lbRequest, defaultResponse)));
			ReflectionUtils.rethrowRuntimeException(exception);
		}
		return null;
	}

	private <T> Object getClientResponse(T response) {
		ClientHttpResponse clientHttpResponse = null;
		if (response instanceof ClientHttpResponse) {
			clientHttpResponse = (ClientHttpResponse) response;
		}
		if (clientHttpResponse != null) {
			try {
				return new ResponseData(clientHttpResponse, null);
			}
			catch (IOException ignored) {
			}
		}
		return response;
	}

	private Set<LoadBalancerLifecycle> getSupportedLifecycleProcessors(String serviceId) {
		return LoadBalancerLifecycleValidator.getSupportedLifecycleProcessors(
				loadBalancerClientFactory.getInstances(serviceId, LoadBalancerLifecycle.class),
				DefaultRequestContext.class, Object.class, ServiceInstance.class);
	}

	@Override
	public URI reconstructURI(ServiceInstance serviceInstance, URI original) {
		return LoadBalancerUriTools.reconstructURI(serviceInstance, original);
	}

	@Override
	public ServiceInstance choose(String serviceId) {
		return choose(serviceId, REQUEST);
	}

	@Override
	public <T> ServiceInstance choose(String serviceId, Request<T> request) {


		/**
		 * spring-cloud-starter-loadbalancer 底层源码流程：
		 * spring-cloud-starter-loadbalancer 提供一个负载均衡工厂 LoadBalancerClientFactory （继承 NamedContextFactory） 用来构建负载均衡客服端
		 *   1. LoadBalancerClientFactory 有个默认配属性类 LoadBalancerClientConfiguration，这个类是每个 远程服务 公共配置！每个 远程服务都有自己独属自己的一份配置，这个配置就是 LoadBalancerClientConfiguration
		 *      每个远程服务配置它是怎么确保配置隔离的？它在远程调用服务时候，如你使用 openfegin 调用远程服务时候，它会在进行负载均衡之前 调用 loadBalancerClientFactory.getInstances(...)方法 传递你的远程服务名称。
		 *      重要的远程服务隔离配置就是在 getInstance 方法里面（此时是进到 LoadBalancerClientFactory 类里面）！ 在方法内部它会 根据 你传递的服务名 创建一个 spring ioc 容器！
		 *      该ioc容器它的配置bean 就是来自于在创建 LoadBalancerClientFactory 类时候指定的 LoadBalancerClientConfiguration 数据配置类；最后把 该ioc 容器添加到 LoadBalancerClientFactory 的 contexts Map里面
		 *      最后进行负载均衡 BlockingLoadBalancerClient.choose() 方法进行负载均衡。
		 *   2. BlockingLoadBalancerClient.choose()
		 *     来的 choose()方法，它会调用 LoadBalancerClientFactory 类 getInstance 方法，根据你的远程服务名称获取一个 loadBalancerClientFactory.getInstance(serviceId) 一个负载均衡器类 ReactiveLoadBalancer。
		 *     注：ReactiveLoadBalancer 这个类就是从上一步的 子容器 创建的
		 *
		 *   3. ReactiveLoadBalancer（实现类 RoundRobinLoadBalancer） 默认实现是轮询
		 *      ReactiveLoadBalancer里面 属性 有一个 ClientFactoryObjectProvider 该作用就是提供一个 ServiceInstanceListSupplier  服务实例对象 ，该对象作用就是去 nacos 获取服务实例 ，最后调用远程服务
		 *      注意： ServiceInstanceListSupplier 这个对象不是真正对外暴露的，它进行了委托，委托在 DelegatingServiceInstanceListSupplier 类 ，所以真正对外暴露的是 DelegatingServiceInstanceListSupplier（实现类 CachingServiceInstanceListSupplier）
		 *     在创建的时候 CachingServiceInstanceListSupplier 类时候，此时就去连接 nacos 服务 获取当前的的服务实例，流程:采用响应式编程方式。
		 *
		 *
		 *  4. 重要对象流程已经关系梳理好了，接下来是梳理调用流程：
		 *     当我们使用 openFegin 调用远程服务。会根据服务名称从 LoadBalancerClientFactory对象  contexts Map 属性获取 spring ioc子容器获取一个负载均衡器对象 RoundRobinLoadBalancer（默认轮询）
		 *     然后获取一个 ServiceInstanceListSupplier 服务实例对 真正对外提供的是 DelegatingServiceInstanceListSupplier （实现类 CachingServiceInstanceListSupplier），此时也知道 在创建 DelegatingServiceInstanceListSupplier的时候，它就通过 真正的 ServiceInstanceListSupplier
		 *     去获取 nacos 上获取当前服务所以服务实例对象，最终采用取模的方式得到最终要调用的服务实例对象。
		 *
		 */
		//根据服务名称id 获取一个响应式的负载均衡器
		ReactiveLoadBalancer<ServiceInstance> loadBalancer = loadBalancerClientFactory.getInstance(serviceId);
		if (loadBalancer == null) {
			return null;
		}

		//进行负载均衡（核心）
		Response<ServiceInstance> loadBalancerResponse = Mono.from(loadBalancer.choose(request)).block();
		if (loadBalancerResponse == null) {
			return null;
		}
		return loadBalancerResponse.getServer();
	}

	private String getHint(String serviceId) {
		LoadBalancerProperties properties = loadBalancerClientFactory.getProperties(serviceId);
		String defaultHint = properties.getHint().getOrDefault("default", "default");
		String hintPropertyValue = properties.getHint().get(serviceId);
		return hintPropertyValue != null ? hintPropertyValue : defaultHint;
	}

}
